package com.zp.qualitypractice.rx;

import io.reactivex.Observable;
import io.reactivex.subjects.PublishSubject;
import io.reactivex.subjects.Subject;

/**
 * Created by izp on 2017/8/13.
 * PublishSubject》Subject》Observable
 */

public class RxbusNoBackpressure {
    private final Subject<Object> mBus;

    private RxbusNoBackpressure() {
        // toSerialized method made bus thread safe
        mBus = PublishSubject.create().toSerialized();
    }

    public static RxbusNoBackpressure get() {
        return Holder.BUS;
    }

    public void post(Object obj) {
        mBus.onNext(obj);
    }

    /**
     * 这个是订阅特定的Object
     * @return
     */
    public <T> Observable<T> toObservable(Class<T> tClass) {
        return mBus.ofType(tClass);
    }


    /**
     * 这个是订阅任意的Objecct
     * @return
     */
    public Observable<Object> toObservable() {
        return mBus;
    }

    public boolean hasObservers() {
        return mBus.hasObservers();
    }

    private static class Holder {
        private static final RxbusNoBackpressure BUS = new RxbusNoBackpressure();
    }

}
